package com.lube.codec;

import java.util.List;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.lube.protobuf.ConnectAckProtos.ConnectAck;
import com.lube.protobuf.ConnectProtos.Connect;
import com.lube.protobuf.PublishAckProtos.PublishAck;
import com.lube.protobuf.PublishProtos.Publish;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageEncoder;

/**
 * 消息编码器
 * <p>
 * 报文协议如下：
 * </p>
 * 
 * <pre>
+--------------+---------+--------------+
|           报文头                           |     数据体               |
+--------------+---------+--------------+
| CommandType  |  Length |    Payload   |
+--------------+---------+--------------+
|      1       |    4    |   ${Length}  |
+--------------+---------+--------------+

1.CommandType：消息指令类型，类型为uint8； 
2.Length：消息体长度，类型为UINT32；
3.Payload: 消息体，使用ProtoBuf压缩。
 * 
 * </pre>
 * 
 * @author caizhiping
 * @date 2017年2月9日 下午2:38:31
 */

@ChannelHandler.Sharable
public class MessageEncoder extends MessageToMessageEncoder<Message> {
	
	private static final Logger LOG = LogManager.getLogger(MessageEncoder.class);
	
	@Override
	protected void encode(ChannelHandlerContext ctx, Message msg, List<Object> out) throws Exception {
		try{
			out.add(doEncode(ctx.alloc(), msg));
		}catch (Exception e) {
			LOG.error(e);
		}
	}

	private ByteBuf doEncode(ByteBufAllocator byteBufAllocator, Message msg) {

		CommandType commandType = msg.getCommandType();

		switch (commandType) {
		case CONNECT:
			return packet(byteBufAllocator, msg.getCommandType().getValue(),
					((Connect) msg.getPayload()).toByteArray());

		case CONNECTACK:
			return packet(byteBufAllocator, msg.getCommandType().getValue(),
					((ConnectAck) msg.getPayload()).toByteArray());

		case PING:
			return packet(byteBufAllocator, msg.getCommandType().getValue(), null);

		case PINGACK:
			return packet(byteBufAllocator, msg.getCommandType().getValue(), null);

		case PUBLISH:
			return packet(byteBufAllocator, msg.getCommandType().getValue(),
					((Publish) msg.getPayload()).toByteArray());

		case PUBACK:
			return packet(byteBufAllocator, msg.getCommandType().getValue(),
					((PublishAck) msg.getPayload()).toByteArray());

		case DISCONNECT:
			return packet(byteBufAllocator, msg.getCommandType().getValue(), null);
		}

		throw new IllegalArgumentException("Unknown message type:" + commandType.getValue());
	}

	private ByteBuf packet(ByteBufAllocator byteBufAllocator, int commandType, byte[] playload) {
		int bodyLength = 0;
		if (playload != null) {
			bodyLength = playload.length;
		}

		ByteBuf data = byteBufAllocator.buffer(bodyLength + MessageDecoder.MAX_HEADER_BYTES);

		data.writeByte(commandType);
		data.writeInt(bodyLength);
		if (playload != null) {
			data.writeBytes(playload);

		}
		return data;
	}
}
